package com.zh.rocketmq;

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;

/**
 * @author Zhou
 * @date 16:25 2025/10/16
 */
public class ProducerExample {
    public static void main(String[] args) throws Exception {
        // 接入点地址，需要设置成Proxy的地址和端口列表
        // 注意：当RocketMQ部署在WSL中时，不能使用127.0.0.1，需要使用WSL的实际IP地址
        var endpoint = "172.24.103.123:8081";

        ClientConfiguration configuration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoint)
                .build();

        ClientServiceProvider provider = ClientServiceProvider.loadService();

        // 消息发送的目标Topic名称，需要提前创建。
        //TopicConfig [topicName=DemoTopic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={}]
        var topic = "DemoTopic";

        // 初始化Producer时需要设置通信配置以及预绑定的Topic。
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();

        for (int i=0;i<5;i++){
            try {
                // 普通消息发送
                Message message = provider.newMessageBuilder()
                        .setTopic(topic)
                        // 设置消息索引键，可根据关键字精确查找某条消息。
                        .setKeys("messageKey")
                        // 设置消息Tag，用于消费端根据指定Tag过滤消息。
                        .setTag("messageTag")
                        // 消息体。
                        .setBody("messageBody".getBytes())
                        .build();

                SendReceipt sendReceipt = producer.send(message);
                System.out.println("Send result: " + sendReceipt);
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println("Message sending failed.");
            }
            Thread.sleep(1000);
        }

        System.out.println("结束...");
        // 关闭生产者
        producer.close();

    }
}
